package com.navinfo.platform.producer;

import com.alibaba.fastjson.JSON;
import com.navinfo.opentsp.platform.location.protocol.common.LCIncidentAlarmData;
//import com.navinfo.platform.bean.incidentAlarmData;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * Created by chen on 2019/7/23.
 */
public class Producer_d009 {
    public static String topic = "incidentAlermPB";

    public static void main(String[] args) throws InterruptedException {
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "platform-010-030-050-128:12892");
        p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
        KafkaProducer<String,byte[]> kafkaProducer = new KafkaProducer(p);

        try {
            long time = 1566061200000L;
            LCIncidentAlarmData.IncidentAlarmData.Builder incidentAlarmData = LCIncidentAlarmData.IncidentAlarmData.newBuilder();
            incidentAlarmData.setTerminalId(13800138009L);
            incidentAlarmData.setSerialNumber(123);
            incidentAlarmData.setReceiveDate(1566061204L);
            incidentAlarmData.setGpsTime(1566061204L);
            incidentAlarmData.setLatitude(110000000);
            incidentAlarmData.setLongitude(22000000);
            incidentAlarmData.setOriginalLat(110000000);
            incidentAlarmData.setOriginalLng(22000000);
            incidentAlarmData.setHeight(100);
            incidentAlarmData.setDirection(123);
            incidentAlarmData.setDepotLogo(123);
            LCIncidentAlarmData.RapidAcceleration.Builder rapidAcceleration = LCIncidentAlarmData.RapidAcceleration.newBuilder();
            rapidAcceleration.setDuration(123);
            rapidAcceleration.setRapidAccelerationFuelConsumption(124);
            rapidAcceleration.setRapidStatus(125);
            incidentAlarmData.setRapidAcceleration(rapidAcceleration);

//

            for(int i=0;i<100;i++) {

                ProducerRecord<String,byte[]> record = new ProducerRecord<String, byte[]>(topic, incidentAlarmData.build().toByteArray());
                kafkaProducer.send(record);
                System.out.println("消息发送成功:" + i);
                Thread.sleep(5);
            }
        }
        catch (Exception e){
            e.printStackTrace();
        }
        finally {
            kafkaProducer.close();
        }

    }
}
